package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.util.DateTimeUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentInfo;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author: xu
 * @desc: 支付宽表处理程序
 * 程  序：MockDB -> MySQL -> FlinkCDC -> Kafka(ZK) -> BaseDbApp -> Kafka/Phoenix(zk/hdfs/hbase) -> OrderWideApp(Redis) -> Kafka -> PaymentWideApp -> Kafka
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/PaymentWideApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2. 从kafka的主题中读取数据
        // 2.1 声明相关的主题以及消费者组
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String groupId = "payment_wide_app_group";
        // 2.1 读取支付数据
        DataStreamSource<String> paymentInfoStrStream = env.addSource(MyKafkaUtil.getKafkaSource(paymentInfoSourceTopic, groupId));
        // 2.2 读取订单宽表数据
        DataStreamSource<String> orderWideStrStream = env.addSource(MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId));

        // TODO 3. 对读取到的数据进行结构的转换，jsonStr->POJO，设置Watermark，和对数据进行分组
        // 3.1 转换支付流和Watermark、分组
        KeyedStream<PaymentInfo, Long> paymentInfoKeyedStream = paymentInfoStrStream
                .map(jsonStr -> JSON.parseObject(jsonStr, PaymentInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((paymentInfo, recordTimestamp) ->
                                        // 需要将字符串的时间转换为毫秒数
                                        DateTimeUtil.stringToTimeStamp(paymentInfo.getCallback_time())
                                )
                )
                .keyBy(PaymentInfo::getOrder_id);
        // 3.2 转换订单流和Watermark、分组
        KeyedStream<OrderWide, Long> orderWideKeyedStream = orderWideStrStream
                .map(jsonStr -> JSON.parseObject(jsonStr, OrderWide.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((orderWide, recordTimestamp) ->
                                        DateTimeUtil.stringToTimeStamp(orderWide.getCreate_time())
                                )
                )
                .keyBy(OrderWide::getOrder_id);
        // paymentInfoKeyedStream.print("paymentInfo>>>>>");
        // orderWideKeyedStream.print("orderWide>>>>");

        // TODO 4.使用IntervalJoin关联两条流
        SingleOutputStreamOperator<PaymentWide> paymentWideStream = paymentInfoKeyedStream
                .intervalJoin(orderWideKeyedStream)
                .between(Time.seconds(-1800), Time.seconds(0))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                             @Override
                             public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, Context ctx, Collector<PaymentWide> out) throws Exception {
                                 out.collect(new PaymentWide(paymentInfo, orderWide));
                             }
                         }
                );
        paymentWideStream.print(">>>>");

        // TODO 5.将数据写到kafka的dwm层
        String paymentWideSinkTopic = "dwm_payment_wide";
        paymentWideStream
                .map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(paymentWideSinkTopic));

        env.execute(PaymentWideApp.class.getSimpleName());
    }
}
